package broker;

import java.net.MalformedURLException;
import java.net.SocketException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;

import broker_manager.BrokerManager;
import client.Client;

import common.BrokerRef;
import common.ClientRef;
import common.DefaultRMISecurityManager;
import common.Link;
import common.LinkComparatorByTo;
import common.LinkListComparatorByFrom;
import common.NetworkParameters;
import common.PublishInfo;
import common.Settings;
import common.Subscription;
import common.SubscriptionInfo;
import common.URIFormatter;
import common.VectorClock;
import common.SubscriptionInfo.Action;

public class BrokerImpl extends UnicastRemoteObject implements Broker {

	public enum BrokerEventType {
		COMING, LEAVING;
	}

	public class PublishDispatcher extends Thread {

		// This variable is volatile because will be used by multiple threads
		private volatile boolean threadSuspended = false;

		public PublishDispatcher() {
			super();
			setName("publish dispatcher");
		}

		@Override
		public void run() {

			// Keep continuously checking the publish queue
			while (true) {
				synchronized (publishQueue) {

					// Deadlock-free thread pause construct
					if (threadSuspended) {
						synchronized (publishDispatcher) {
							while (threadSuspended)
								try {
									publishDispatcher.wait();
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
						}
					}

					System.out.print("Publish dispatcher: checking...");

					// If an element is found, handle it
					if (!publishQueue.isEmpty()) {

						// Get the first element in the queue
						PublishInfo firstElement = publishQueue.iterator()
								.next();

						// If the element going to be processed is marked,
						// meaning a complete queue loop has been performed,
						// stop cycling
						if (firstElement.requeued) {
							System.out
									.println(" waiting (queue infinite looping prevention)");
							try {
								synchronized (publishDispatcher) {
									publishDispatcher.wait();
								}
							} catch (InterruptedException e) {
								e.printStackTrace();
								return;
							}
						}

						// No problems found, keep on processing
						else {
							System.out.println(" element found");
							try {
								processPublish(publishQueue.poll());
							} catch (RemoteException e) {
								e.printStackTrace();
								return;
							} catch (Exception e) {
								e.printStackTrace();
								return;
							}
						}
					}

					// If the queue is empty, wait
					else {
						System.out.println(" waiting (empty queue)");
						try {
							synchronized (publishDispatcher) {
								publishDispatcher.wait();
							}
						} catch (InterruptedException e) {
							e.printStackTrace();
							return;
						}
					}
				}
			}
		}
	}

	public class SubscriptionsDispatcher extends Thread {

		// This variable is volatile because will be used by multiple threads
		private volatile boolean threadSuspended = false;

		public SubscriptionsDispatcher() {
			super();
			setName("subscriptions dispatcher");
		}

		@Override
		public void run() {

			// Keep continuously checking the subscriptions queue
			while (true) {
				synchronized (subscriptionsQueue) {

					// Deadlock-free thread pause construct
					if (threadSuspended) {
						synchronized (subscriptionsDispatcher) {
							while (threadSuspended)
								try {
									subscriptionsDispatcher.wait();
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
						}
					}

					System.out.print("Subscription dispatcher: checking...");

					// If an element is found, handle it
					if (!subscriptionsQueue.isEmpty()) {
						System.out.println(" element found");
						try {
							processSubscription(subscriptionsQueue.poll());
						} catch (Exception e) {
							e.printStackTrace();
							return;
						}
					}

					// If the queue is empty, wait
					else {
						System.out.println(" waiting");
						try {
							synchronized (subscriptionsDispatcher) {
								subscriptionsDispatcher.wait();
							}
						} catch (InterruptedException e) {
							e.printStackTrace();
							return;
						}
					}
				}
			}
		}
	}

	public enum SubscriptionType {
		LOCAL, REMOTE;
	}

	private static final long serialVersionUID = 1L;
	private BrokerManager brokerManager;
	private URI brokerManagerURI;
	private List<BrokerRef> brokers = Collections
			.synchronizedList(new ArrayList<BrokerRef>());
	private List<ClientRef> clients = Collections
			.synchronizedList(new ArrayList<ClientRef>());
	private Hashtable<URI, List<Subscription>> localSubscriptionsTable = new Hashtable<URI, List<Subscription>>();
	private URI localURI;
	private PublishDispatcher publishDispatcher = new PublishDispatcher();
	private ConcurrentLinkedQueue<PublishInfo> publishQueue = new ConcurrentLinkedQueue<PublishInfo>();
	private Hashtable<URI, List<Subscription>> remoteSubscriptionsTable = new Hashtable<URI, List<Subscription>>();
	private Hashtable<URI, Hashtable<URI, URI>> routingTable = new Hashtable<URI, Hashtable<URI, URI>>();
	private SubscriptionsDispatcher subscriptionsDispatcher = new SubscriptionsDispatcher();
	private ConcurrentLinkedQueue<SubscriptionInfo> subscriptionsQueue = new ConcurrentLinkedQueue<SubscriptionInfo>();
	private Hashtable<String, VectorClock> topicsVectorClocks = new Hashtable<String, VectorClock>();

	public BrokerImpl() throws RemoteException, UnknownHostException,
			URISyntaxException {
		super();
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new DefaultRMISecurityManager());
		}
	}

	@Override
	public void addClient(URI uri) throws MalformedURLException,
			RemoteException, NotBoundException, Exception {
		if (clients.contains(new ClientRef(uri, null))) {
			throw new Exception("Duplicated client");
		}
		clients.add(new ClientRef(uri, (Client) Naming.lookup("//"
				+ uri.toString() + "/Client")));
		System.out.println("Client " + uri.toString() + " added");
	}

	@Override
	public void beginNetworkReorganization(URI brokerURI,
			BrokerEventType eventType) throws MalformedURLException,
			RemoteException, NotBoundException {

		// New broker case
		if (eventType == BrokerEventType.COMING) {

			// Add the broker to the list
			brokers.add(new BrokerRef(brokerURI, (Broker) Naming.lookup("//"
					+ brokerURI.toString() + "/Broker")));
			System.out.println("Broker " + brokerURI.toString() + " coming");
		}

		// Leaving broker case
		else {

			// Remove the broker from the list
			brokers.remove(new BrokerRef(brokerURI, null));
			System.out.println("Broker " + brokerURI.toString() + " leaving");
		}

		// The network reorganization will be handled by the new broker
		// This will be printed in all the brokers except of the new one because
		// the broker manager does so
		System.out.println("+---------------------------------+");
		System.out.println("| STARTING NETWORK REORGANIZATION |");
		System.out.println("+---------------------------------+");

		// Pause the dispatchers

		// Deadlock-free thread pause
		// Note that the method pause() id deprecated because deadlock prone
		synchronized (subscriptionsDispatcher) {
			subscriptionsDispatcher.threadSuspended = true;

			// If the thread was already waiting because of empty queue, this
			// notify() lets the tread wait in the "threadSuspended" loop
			subscriptionsDispatcher.notify();
		}
		System.out.println("Subscriptions dispatcher thread suspended");

		// Deadlock-free thread pause
		// Note that the method pause() id deprecated because deadlock prone
		synchronized (publishDispatcher) {
			publishDispatcher.threadSuspended = true;

			// If the thread was already waiting because of empty queue, this
			// notify() lets the tread wait in the "threadSuspended" loop
			publishDispatcher.notify();
		}
		System.out.println("Publish dispatcher thread suspended");

		if (eventType == BrokerEventType.LEAVING) {

			// Remove broker's clients' subscriptions from the remote
			// subscriptions table
			System.out.println("Clearing " + brokerURI
					+ "'s clients' subscriptions");
			remoteSubscriptionsTable.remove(brokerURI);
		}

		// Clear the routing table
		routingTable = new Hashtable<URI, Hashtable<URI, URI>>();
		System.out.println("ROUTING TABLE CLEARED");
	}

	@Override
	public void endNetworkReorganization(URI brokerURI,
			BrokerEventType eventType) {
		System.out.println("+----------------------------------+");
		System.out.println("| NETWORK REORGANIZATION COMPLETED |");
		System.out.println("+----------------------------------+");

		// Print the new routing table
		System.out.println("UPDATED ROUTING TABLE");
		Enumeration<URI> e1 = routingTable.keys(), e2;
		Hashtable<URI, URI> ht = new Hashtable<URI, URI>();
		URI key1, key2;
		System.out.println("      FROM       |       TO        |    NEXT HOP");
		while (e1.hasMoreElements()) {
			key1 = e1.nextElement();
			ht = routingTable.get(key1);
			e2 = ht.keys();
			while (e2.hasMoreElements()) {
				key2 = e2.nextElement();
				System.out.println(" " + URIFormatter.format(key1) + " | "
						+ URIFormatter.format(key2) + " | "
						+ URIFormatter.format(ht.get(key2)));
			}
		}

		// Reorganize the subscriptions in the queue setting the local uri
		// as the source to guarantee that the topology change doesn't cause
		// any subscription to be lost
		for (SubscriptionInfo subInfo : subscriptionsQueue)
			subInfo.source = localURI;

		// Same for publish but also update vectorClocks contained in queued
		// publishInfos
		for (PublishInfo pubInfo : publishQueue) {
			pubInfo.source = localURI;

			// If a broker was coming, add a clock to the vector
			// clock contained in each pubInfo queued locally
			if (eventType == BrokerEventType.COMING) {

				// This must not be done on pubInfo in the queue coming from
				// attached clients, because they haven't yet received their
				// vector clock
				if (pubInfo.vectorClock != null)

					// This method adds the clock only if not already contained
					pubInfo.vectorClock.addClock(brokerURI, 0);
			}

			// If a broker was leaving, remove its clock from the vector
			// clock contained in each pubInfo queued locally
			else {

				// This must not be done on pubInfo in the queue coming from
				// attached clients, because they haven't yet received their
				// vector clock
				if (pubInfo.vectorClock != null)
					pubInfo.vectorClock.removeClock(brokerURI);
			}
		}

		// Update topicVectorClocks adding the clock for the coming broker
		if (eventType == BrokerEventType.COMING) {

			// Add the broker to topicsVectorClocks
			Enumeration<String> e = topicsVectorClocks.keys();
			String topic;
			while (e.hasMoreElements()) {
				topic = e.nextElement();
				topicsVectorClocks.get(topic).addClock(brokerURI, 0);
			}
		}

		// Update topicVectorClocks removing the clock for the leaving broker
		else {

			// Remove the broker from topicsVectorClocks
			Enumeration<String> e = topicsVectorClocks.keys();
			String topic;
			while (e.hasMoreElements()) {
				topic = e.nextElement();
				topicsVectorClocks.get(topic).removeClock(brokerURI);
			}
		}

		// Resume the subscriptions dispatcher thread
		// Note that the method resume() is deprecated
		// May be that the dispatcher wasn't running if this method is
		// executed on a new broker joining the network
		if (!subscriptionsDispatcher.isAlive()) {
			subscriptionsDispatcher.start();
			System.out.println("Subscriptions dispatcher thread started");
		} else {
			synchronized (subscriptionsDispatcher) {
				subscriptionsDispatcher.threadSuspended = false;
				subscriptionsDispatcher.notify();
			}
			System.out.println("Subscriptions dispatcher thread resumed");
		}

		// Resume the publish dispatcher thread
		// Note that the method resume() is deprecated
		// May be that the dispatcher wasn't running if this method is
		// executed
		// on a new broker joining the network
		if (!publishDispatcher.isAlive()) {
			publishDispatcher.start();
			System.out.println("Publish dispatcher thread started");
		} else {
			synchronized (publishDispatcher) {
				publishDispatcher.threadSuspended = false;
				publishDispatcher.notify();
			}
			System.out.println("Publish dispatcher thread resumed");
		}
	}

	private BrokerRef getBrokerRefByURI(URI uri) {
		for (BrokerRef br : brokers)
			if (br.uri.equals(uri))
				return br;
		return null;
	}

	@Override
	public ArrayList<URI> getBrokers() {
		ArrayList<URI> list = new ArrayList<URI>();
		for (BrokerRef br : brokers) {
			list.add(br.uri);
		}
		return list;
	}

	private ClientRef getClientRefByURI(URI uri) {
		for (ClientRef cr : clients)
			if (cr.uri.equals(uri))
				return cr;
		return null;
	}

	private Link getLinkByDestination(ArrayList<Link> links, URI uri) {
		for (Link l : links) {
			if (l.to.equals(uri))
				return l;
		}
		return null;
	}

	private ArrayList<Link> getLinksBySource(
			ArrayList<ArrayList<Link>> bidirNetMap, URI uri) {
		for (ArrayList<Link> list : bidirNetMap) {
			if (list.get(0).from.equals(uri))
				return list;
		}
		return null;
	}

	@Override
	public ArrayList<Link> getLinkState() {
		ArrayList<Link> links = new ArrayList<Link>();
		long lat[] = new long[NetworkParameters.numPings];
		long start, end, sum;
		System.out.println("Local link state:");
		for (BrokerRef br : brokers) {
			System.out.print(" " + localURI + " --> " + br.uri + " -");
			try {
				sum = 0;
				if (!br.uri.equals(localURI)) {
					for (int i = 0; i < lat.length; i++) {
						start = System.currentTimeMillis();
						br.broker.ping();
						end = System.currentTimeMillis();
						lat[i] = end - start;
						System.out.print(" " + lat[i] + "ms");
						sum += lat[i];
					}
				}
				System.out.println(" (ave: " + sum / lat.length + "ms)");
				links.add(new Link(localURI, br.uri, sum / lat.length));
			} catch (RemoteException e) {
				System.out
						.println(" Link may be broken, latency set to infinity");
				links.add(new Link(localURI, br.uri, Long.MAX_VALUE));
			}
		}
		return links;
	}

	private URI getNextHop(URI to, ArrayList<Link> spt) {
		Link l = getLinkByDestination(spt, to);
		if (l == null)
			return null;
		else if (l.from.equals(localURI))
			return to;
		else
			return getNextHop(l.from, spt);
	}

	@Override
	public Hashtable<URI, List<Subscription>> getRemoteSubscriptionsTable() {
		return remoteSubscriptionsTable;
	}

	@Override
	public Hashtable<String, VectorClock> getTopicsVectorClocks()
			throws RemoteException {
		return topicsVectorClocks;
	}

	private void organizeNetworkMap(ArrayList<ArrayList<Link>> networkMap) {

		// Sort each row by field "to" of "link"
		for (ArrayList<Link> list : networkMap)
			Collections.sort(list, new LinkComparatorByTo());

		// Sort rows by field "from" of "link"
		Collections.sort(networkMap, new LinkListComparatorByFrom());
	}

	@Override
	public void ping() {

		// Increase latency variance
		// Effectiveness of this operation increases if numPings in very low
		try {
			Thread.sleep(new Random().nextInt(10) * 2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return;
	}

	private void printSubscriptionsTable(SubscriptionType type) {

		// Get the correct table
		Hashtable<URI, List<Subscription>> subscriptionsTable;
		if (type == SubscriptionType.LOCAL) {
			subscriptionsTable = localSubscriptionsTable;
		} else {
			subscriptionsTable = remoteSubscriptionsTable;
		}

		// Print subscription table
		System.out.println(type.toString().toUpperCase()
				+ " SUBSCRIPTION TABLE");
		Enumeration<URI> e1 = subscriptionsTable.keys();
		URI key;
		List<Subscription> list;
		if (!e1.hasMoreElements()) {
			System.out.println(" no entries");
		}
		while (e1.hasMoreElements()) {
			key = e1.nextElement();
			System.out.print(" " + key);
			list = subscriptionsTable.get(key);
			for (Subscription sub : list)
				System.out.print(" " + sub.topic + "(" + sub.numSubscribers
						+ ")");
			System.out.println();
		}
	}

	private void processPublish(PublishInfo pubInfo) throws Exception {

		synchronized (Settings.sleepTime) {
			Thread.sleep(Settings.sleepTime);
		}

		// If the pubInfo carries a not initialized vector clocks, it is coming
		// from an attached client
		boolean isFromClient = pubInfo.vectorClock == null;

		if (isFromClient) {
			System.out.print("Publish \"" + pubInfo.msg + "\" from client "
					+ pubInfo.source + " (" + pubInfo.topic + ") ");
		} else {
			System.out.print("Publish \"" + pubInfo.msg + "\" from broker "
					+ pubInfo.source + " (" + pubInfo.topic + ") ");
			System.out.println(pubInfo.vectorClock.toString());
		}

		System.out.println("Publish queue lenght: " + publishQueue.size());

		// Get thread-safely the entropy value
		float entropy;
		synchronized (Settings.queueEntropy) {
			entropy = Settings.queueEntropy;
		}

		// If the publish has been called by a client, initialize the
		// vectorClock field
		if (isFromClient) {
			VectorClock vc = topicsVectorClocks.get(pubInfo.topic);
			if (vc != null) {
				pubInfo.vectorClock = vc.getCopy();
				pubInfo.vectorClock.incrementClock(localURI);
				System.out.println(pubInfo.vectorClock.toString());
			} else {
				System.out
						.println("Unknown vector clock, waiting for the proper subscription to initialize it");
				for (PublishInfo pi : publishQueue)
					pi.requeued = false;
				publishQueue.offer(pubInfo);
				publishQueue.notify();
				return;
			}
		} else if (new Random().nextFloat() < entropy) {

			// Queue entropy generation mechanism, used to simulate non-FIFO
			// channels
			System.out.println("Message \"" + pubInfo.msg + "\" ["
					+ pubInfo.topic + "] requeued (entropy generator)");

			// Remove the mark from all the publish messages in the queue
			// because the acceptance of the last message could have resolved a
			// causal waiting in any topic
			for (PublishInfo pi : publishQueue)
				pi.requeued = false;
			publishQueue.offer(pubInfo);
			publishQueue.notify();
			return;
		}

		// Compare vectorClocks
		VectorClock vc = topicsVectorClocks.get(pubInfo.topic);
		if (!vc.comparable(pubInfo.vectorClock)) {

			// This shouldn't happen: if it does, it is a fatal error!
			System.out.println("ERROR: Uncomparable vector clocks!");
			throw new Exception("Uncomparable vector clocks!");
		}

		boolean unacceptable = false, greaterFound = false;

		// If this is not the first message received i need to perform usual
		// checks on acceptability, else the message is accepted.
		if (vc.isInitialized()) {
			for (int i = 0; i < vc.clocksNumber(); i++) {

				// If a clock in the pubInfo is greater than the local clock
				if (vc.clockValue(i) < pubInfo.vectorClock.clockValue(i)) {

					// If it is greater by 1
					if (vc.clockValue(i) + 1 == pubInfo.vectorClock
							.clockValue(i)) {

						// If this is the first greater clock found
						if (!greaterFound)
							greaterFound = true;

						// If this isn't the only greater clock, drop
						else {
							System.out
									.println("Missing some previous message/s from other broker/s");
							unacceptable = true;
						}
						break;

					}

					// If is greater by more than 1, drop
					else {
						System.out.println("Missing some previous message/s");
						unacceptable = true;
						break;
					}
				}
			}
		} else {
			System.out.println("Vector clock for topic " + pubInfo.topic
					+ " was not initialized");

			// Even if no greater clock has been neither searched for, being
			// this topic's clock uninitialized, the publish info received
			// certainly carries new infos.
			greaterFound = true;
		}

		// If marked as unacceptable, put the pubInfo at the tail of the queue
		if (unacceptable) {
			System.out.println("Re-queueing message");

			// Mark this message as already processed but found unacceptable
			pubInfo.requeued = true;

			// Actually re-queue
			publishQueue.offer(pubInfo);
		} else {

			// Remove the mark from all the publish messages in the queue
			// because the acceptance of the last message could have resolved a
			// causal waiting in any topic
			for (PublishInfo pi : publishQueue)
				pi.requeued = false;

			ArrayList<URI> forwarding = new ArrayList<URI>();
			List<Subscription> subscriptions;
			Enumeration<URI> e;
			URI key;

			// Process the message locally only if it carries some new info
			if (greaterFound) {

				// Set the local vector clock, corresponding to the received
				// message's topic, to the one contained in the received
				// message.
				VectorClock newVc = pubInfo.vectorClock.getCopy();
				newVc.mergeWith(vc);
				if(!isFromClient) {
					newVc.setInitialized(true);
				}
				topicsVectorClocks.put(pubInfo.topic, newVc);

				// Check in local subscriptions table for attached subscribed
				// clients
				e = localSubscriptionsTable.keys();
				while (e.hasMoreElements()) {
					key = e.nextElement();
					subscriptions = localSubscriptionsTable.get(key);
					for (Subscription sub : subscriptions) {
						if (sub.topic.equals(pubInfo.topic)) {
							forwarding.add(key);
							break;
						}
					}
				}

				// Print the message to locally subscribed clients
				for (URI uri : forwarding) {
					if (getClientRefByURI(uri) != null) {
						getClientRefByURI(uri).client.printMessage(
								pubInfo.username, pubInfo.msg, pubInfo.topic);
						System.out.println(" --> printed to client " + uri);
					}
				}
			}

			// This must be done also for "old" info, because other broker may
			// still need to process that
			// Check in remote subscriptions table for remotely subscribed
			// brokers
			forwarding.clear();
			e = remoteSubscriptionsTable.keys();
			while (e.hasMoreElements()) {
				key = e.nextElement();
				subscriptions = remoteSubscriptionsTable.get(key);
				for (Subscription sub : subscriptions) {
					if (sub.topic.equals(pubInfo.topic)) {
						forwarding.add(key);
						break;
					}
				}
			}

			// Forward the message to the proper next hop brokers
			HashSet<URI> nextHops = new HashSet<URI>();
			URI nextHop;
			for (URI destURI : forwarding) {
				if (routingTable.get(pubInfo.source) != null) {
					nextHop = routingTable.get(pubInfo.source).get(destURI);
					if (nextHop != null)
						nextHops.add(nextHop);
				}
			}
			if (nextHops.size() > 0) {
				System.out.print(" --> published to brokers:");
				for (URI uri : nextHops) {
					getBrokerRefByURI(uri).broker.publish(pubInfo);
					System.out.print(" " + uri);
				}
				System.out.println();
			}
		}
	}

	private void processSubscription(SubscriptionInfo info) throws Exception {

		synchronized (Settings.sleepTime) {
			Thread.sleep(Settings.sleepTime);
		}

		// Check if this message has already been processed
		boolean visited = info.isVisited(localURI);

		if (info.action == SubscriptionInfo.Action.ADD)
			System.out.println("Processing subscribe message received from "
					+ info.source + " (subscriber: " + info.subscriber
					+ ", topic: " + info.topic + ")");
		else
			System.out.println("Processing unsubscribe message received from "
					+ info.source + " (unsubscriber: " + info.subscriber
					+ ", topic: " + info.topic + ")");

		System.out.println("Subscription queue lenght: "
				+ subscriptionsQueue.size());

		// If the subscription comes from one of your associated clients...
		// The clientStopping is added because a client stopping may have
		// already been removed from the clients list, but its unsubscribe
		// messages still need processing.
		if (clients.contains(new ClientRef(info.subscriber, null))
				|| info.clientStopping) {

			// ...update local subscriptions table only if this info has not
			// been
			// processed here previously
			if (!visited) {
				updateSubscriptionsTable(info, SubscriptionType.LOCAL, false);

				// Print local subscriptions table
				printSubscriptionsTable(SubscriptionType.LOCAL);
			}

			// Edit the subscription setting local uri as subscriber uri and
			// source uri
			info.source = localURI;
			info.subscriber = localURI;

			// Other brokers don't need to look in the local subscription table
			info.clientStopping = false;
		}

		// In any case, process the subscription as remote

		// Update remote subscriptions table only if this info has not been
		// processed here previously
		if (!visited) {
			updateSubscriptionsTable(info, SubscriptionType.REMOTE, true);

			// Print remote subscriptions table
			printSubscriptionsTable(SubscriptionType.REMOTE);

			// Mark this info as already processed
			info.addVisited(localURI);
		} else {
			System.out.println("This subscription has already been processed");
		}

		// Forward the message to the proper next hop brokers
		Hashtable<URI, URI> ht = routingTable.get(info.source);
		if (ht != null) {
			Enumeration<URI> e2 = ht.keys();
			HashSet<URI> nextHops = new HashSet<URI>();
			URI dest;
			BrokerRef nhbr;
			while (e2.hasMoreElements()) {
				dest = e2.nextElement();
				nhbr = getBrokerRefByURI(ht.get(dest));
				nextHops.add(nhbr.uri);
			}
			if (nextHops.size() > 0)
				System.out.print(" --> forwarded to:");
			for (URI uri : nextHops) {
				getBrokerRefByURI(uri).broker.subscribe(info);
				System.out.print(" " + uri);
			}
			if (nextHops.size() > 0)
				System.out.println();
		}
	}

	@Override
	public void publish(PublishInfo pubInfo) throws RemoteException {

		// Put publish into the queue
		publishQueue.offer(pubInfo);

		// Signal the eventually waiting publish dispatcher
		synchronized (publishDispatcher) {
			publishDispatcher.notify();
		}
	}

	@Override
	public void removeClient(URI uri) throws Exception, RemoteException {
		if (!clients.contains(new ClientRef(uri, null))) {
			throw new Exception("Not registered client");
		}
		System.out.println("Unregister request received from " + uri);

		// Unsubscribe client's topics
		System.out.println("Forwarding proper unsubscribe messages");
		List<Subscription> subs = localSubscriptionsTable.get(uri);
		ArrayList<SubscriptionInfo> subInfos = new ArrayList<SubscriptionInfo>();
		if (subs != null) {

			// Create the unsubscribe requests
			for (Subscription s : subs) {
				SubscriptionInfo tempSubInfo = new SubscriptionInfo(localURI,
						uri, s.topic, SubscriptionInfo.Action.REMOVE, true);
				tempSubInfo.addVisited(localURI);
				subInfos.add(tempSubInfo);
			}

			// Put the requests into the queue
			// This two-steps procedure is necessary to avoid concurrent
			// modification of "subs"
			for (SubscriptionInfo subInfo : subInfos) {

				// Put subscription into the queue
				subscriptionsQueue.offer(subInfo);

				// Signal the eventually waiting subscriptions dispatcher
				synchronized (subscriptionsDispatcher) {
					subscriptionsDispatcher.notify();
				}
			}
		}

		// Actually remove the client
		clients.remove(new ClientRef(uri, null));
		System.out.println("Client " + uri.toString() + " removed");
	}

	public void start(String networkInterface, URI brokerManagerURI)
			throws MalformedURLException, RemoteException, NotBoundException,
			Exception, SocketException {

		// Setting up connection with broker manager
		this.brokerManagerURI = brokerManagerURI;
		localURI = NetworkParameters.getLocalURI(networkInterface);
		Naming.rebind("Broker", this);
		System.out.println("Broker started on " + localURI);
		brokerManager = (BrokerManager) Naming.lookup("//"
				+ this.brokerManagerURI.toString() + "/BrokerManager");
		System.out.println("Connected to broker manager "
				+ this.brokerManagerURI.toString());
		System.out.println("Registering as " + localURI);

		// Register to broker manager
		ArrayList<URI> brokersURIs = brokerManager.registerBroker(localURI);
		while (brokersURIs == null) {
			System.out.println("The network is busy updating, waiting "
					+ NetworkParameters.backoffPeriod + " seconds");
			try {
				Thread.sleep(NetworkParameters.backoffPeriod * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			brokersURIs = brokerManager.registerBroker(localURI);
		}

		// After the method register broker has been called, ALL brokers have
		// empty routing tables and the complete list of brokers (including the
		// new one)

		// Fill brokers list with remote references
		for (URI uri : brokersURIs) {
			brokers.add(new BrokerRef(uri, (Broker) Naming.lookup("//"
					+ uri.toString() + "/Broker")));
			System.out.print("Broker " + uri.toString() + " added");
			if (uri.equals(localURI))
				System.out.println(" (localhost)");
			else
				System.out.println();
		}

		// Start the network reorganization procedure
		updateNetwork(localURI, BrokerEventType.COMING, true);
	}

	public void stop() throws RemoteException, MalformedURLException,
			NotBoundException, Exception {

		// Unregister from broker manager
		System.out.println("Shutting down...");
		boolean success = brokerManager.unregisterBroker(localURI);
		while (!success) {
			System.out.println("The network is busy updating, waiting "
					+ NetworkParameters.backoffPeriod + " seconds");
			try {
				Thread.sleep(NetworkParameters.backoffPeriod * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			success = brokerManager.unregisterBroker(localURI);
		}

		// Notifying attached clients
		System.out.println("Notifying attached clients");
		for (ClientRef cr : clients) {
			System.out.println(" --> " + cr.uri);
			cr.client.notifyBrokerShutdown();
		}

		// At this point, no more publish or subscribe requests can come from
		// associated clients

		// Be sure to have handled all the subscription and publish messages in
		// the queues so that no messages get lost
		while (!subscriptionsQueue.isEmpty()) {
			System.out.println("Subscription queue isn't empty, waiting "
					+ NetworkParameters.backoffPeriod + " seconds");
			try {
				Thread.sleep(NetworkParameters.backoffPeriod * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		while (!publishQueue.isEmpty()) {
			System.out.println("Publish queue isn't empty, waiting "
					+ NetworkParameters.backoffPeriod + " seconds");
			try {
				Thread.sleep(NetworkParameters.backoffPeriod * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		// Now anyone is idle waiting for the network reorganization to begin,
		// so pick a random broker and tell him to update the network
		// NB: this broker isn't idle, its dispatchers are still running and its
		// routing table is the old one
		brokers.remove(new BrokerRef(localURI, null));
		if (brokers.size() > 0) {

			// This method causes the topology to be updated
			brokers.get(new Random().nextInt(brokers.size())).broker
					.updateNetwork(localURI, BrokerEventType.LEAVING, false);
		}

		// The topology is now ready to resume its activities

		// Unbinding
		Naming.unbind("Broker");

		// Print something big
		System.out.println("+----------------+");
		System.out.println("| BROKER STOPPED |");
		System.out.println("+----------------+");

		// Kill the process
		Runtime.getRuntime().exit(0);
	}

	@Override
	// Called only by associated clients
	public void subscribe(SubscriptionInfo info) throws RemoteException {

		// Put subscription into the queue
		subscriptionsQueue.offer(info);

		// Signal the eventually waiting subscriptions dispatcher
		synchronized (subscriptionsDispatcher) {
			subscriptionsDispatcher.notify();
		}
	}

	@Override
	public void updateNetwork(URI uri, BrokerEventType eventType,
			boolean getSubscriptionsTable) throws RemoteException {

		// This will be printed only in the broker managing the reorganization
		System.out.println("NETWORK UPDATE REQUEST RECIVED");
		ArrayList<ArrayList<Link>> networkMap = new ArrayList<ArrayList<Link>>();
		ArrayList<Link> linkState;

		// Gather links states
		for (BrokerRef br : brokers) {
			linkState = br.broker.getLinkState();
			if (br.uri.equals(localURI))
				System.out.println("Local link state computed");
			else
				System.out.println("Got remote link state from " + br.uri);
			networkMap.add(linkState);
		}
		organizeNetworkMap(networkMap);

		// Make links bidirectional
		ArrayList<ArrayList<Link>> bidirNetMap = new ArrayList<ArrayList<Link>>();
		long aveLat;
		for (int i = 0; i < networkMap.size(); i++) {
			bidirNetMap.add(new ArrayList<Link>());
			for (int j = 0; j < networkMap.size(); j++) {
				aveLat = (networkMap.get(i).get(j).latency + networkMap.get(j)
						.get(i).latency) / 2;
				bidirNetMap.get(i).add(
						new Link(networkMap.get(i).get(j).from, networkMap.get(
								i).get(j).to, aveLat));
			}
		}

		// Print averaged bidirectional latencies
		System.out.println("BIDIRECTIONAL LATENCIES");
		for (ArrayList<Link> list : bidirNetMap) {
			System.out.println(list.get(0).from);
			for (Link l : list) {
				System.out.println(" --> " + l.to + "   " + l.latency + "ms");
			}
		}

		// Distributed SPT and Routing table update
		for (BrokerRef br : brokers)
			br.broker.updateTopology(bidirNetMap);

		// At this point, each broker have computed his SPT and told anyone else
		// to update their routing tables according to that informations

		// Get up to date remote subscriptions table and topics vector clocks
		// Get it from the first "next hop" entry in the routing table, it will
		// be someone 1-hop away in local SPT
		if (getSubscriptionsTable && !routingTable.get(localURI).isEmpty()) {
			Hashtable<URI, URI> myRoutingTable = routingTable.get(localURI);
			URI anyNextHop = null;
			Enumeration<URI> e = myRoutingTable.keys();
			while (e.hasMoreElements()) {
				anyNextHop = myRoutingTable.get(e.nextElement());
				if (!anyNextHop.equals(localURI))
					break;
			}
			remoteSubscriptionsTable = getBrokerRefByURI(anyNextHop).broker
					.getRemoteSubscriptionsTable();
			System.out.println("Got remote subscription table from "
					+ anyNextHop);
			printSubscriptionsTable(SubscriptionType.REMOTE);
		}

		// The system is ready to resume dispatchers activities, notify other
		// brokers
		for (BrokerRef br : brokers)
			br.broker.endNetworkReorganization(uri, eventType);

		// Notify the broker manager that everything is running normally again
		brokerManager.networkUpdateComplete();

		// Print something very visible
		System.out.println("NETWORK UPDATE REQUEST COMPLETED");
	}

	@Override
	public void updateRoutingTable(ArrayList<Link> spt, URI source) {

		// Update the routing table by adding routing information obtained from
		// "source"'s "spt"
		System.out.println("Updating routing table with " + source + " SPT");
		Hashtable<URI, URI> ht = new Hashtable<URI, URI>();
		URI nextHop;
		routingTable.put(source, ht);

		// For each destination broker...
		for (BrokerRef br : brokers) {

			// ...compute next hop according to received spt (from "source" to
			// "br")
			nextHop = getNextHop(br.uri, spt);
			if (nextHop != null)
				ht.put(br.uri, nextHop);
		}
	}

	@Override
	public void updateSubscriptionsTable(SubscriptionInfo info,
			SubscriptionType type, boolean updateVectorClocks) {

		// Get the correct table
		Hashtable<URI, List<Subscription>> subscriptionsTable;
		if (type == SubscriptionType.LOCAL)
			subscriptionsTable = localSubscriptionsTable;
		else
			subscriptionsTable = remoteSubscriptionsTable;

		// Get the correct subscriptions list from the table
		List<Subscription> subs = subscriptionsTable.get(info.subscriber);

		// Subscription case
		if (info.action == Action.ADD) {

			// First registration from the source uri for any topic
			if (subs == null) {
				List<Subscription> newSub = Collections
						.synchronizedList(new ArrayList<Subscription>());
				newSub.add(new Subscription(info.topic, 1));
				subscriptionsTable.put(info.subscriber, newSub);
			} else {

				// Already present subscription
				boolean found = false;
				for (Subscription sub : subs) {
					if (sub.topic.equals(info.topic)) {
						sub.numSubscribers++;
						found = true;
					}
				}

				// New topic subscription from already present source
				if (!found)
					subs.add(new Subscription(info.topic, 1));
			}

			// Management completed
			if (type == SubscriptionType.LOCAL)
				System.out.println("Local subscription managed");
			else
				System.out.println("Remote subscription managed");
		}

		// Unsubscription case
		else {

			// If for some reason an unsubscription message is processed before
			// its "corresponding" subscribe message, requeue it
			if (subs == null) {
				subscriptionsQueue.offer(info);
				System.out
						.println("Unsubscribe message received out of order, requeueing");
				return;
			}

			// Find the right topic
			boolean found = false;
			for (Subscription s : subs) {
				if (s.topic.equals(info.topic)) {
					found = true;

					// If for some reason an unsubscription message is processed
					// before
					// its "corresponding" subscribe message, requeue it
					if (s.numSubscribers == 0) {
						subscriptionsQueue.offer(info);
						System.out
								.println("Unsubscribe message received out of order, requeueing");
						return;
					}

					// Decrement the number of subscribed hosts...
					s.numSubscribers--;

					// ...if it becomes zero, remove the entry
					if (s.numSubscribers <= 0) {
						subs.remove(s);
						if (topicsVectorClocks.get(info.topic) != null)
							topicsVectorClocks.get(info.topic).setInitialized(
									false);
						if (subs.size() <= 0) {
							subscriptionsTable.remove(info.subscriber);
						}
					}
					break;
				}
			}

			// If for some reason an unsubscription message is processed before
			// its "corresponding" subscribe message, requeue it
			if (!found) {
				subscriptionsQueue.offer(info);
				System.out
						.println("Unsubscribe message received out of order, requeueing");
				return;
			}

			// Management completed
			if (type == SubscriptionType.LOCAL)
				System.out.println("Local unsubscription managed");
			else
				System.out.println("Remote unsubscription managed");
		}

		// Eventually update
		if (updateVectorClocks)
			updateTopicVectorClocks(info);
	}

	@Override
	public void updateTopicVectorClocks(SubscriptionInfo info) {

		// Subscription case
		if (info.action == Action.ADD) {

			// Insert the new topic row
			if (!topicsVectorClocks.containsKey(info.topic)) {
				VectorClock vc = new VectorClock();
				for (BrokerRef br : brokers)
					vc.addClock(br.uri, 0);
				topicsVectorClocks.put(info.topic, vc);
			}
		}

		// Unsubscription case
		else {

			// Check if someone is still registered to that topic: if not,
			// remove that topic from the topicsVectorClocks
			boolean remove = true;
			Enumeration<URI> e = remoteSubscriptionsTable.keys();
			URI uri;
			List<Subscription> subs;
			while (e.hasMoreElements()) {
				uri = e.nextElement();
				subs = remoteSubscriptionsTable.get(uri);
				if (subs.contains(new Subscription(info.topic, 0))) {
					remove = false;
					break;
				}
			}
			if (remove) {
				topicsVectorClocks.remove(info.topic);
			}
		}
	}

	@Override
	public void updateTopology(ArrayList<ArrayList<Link>> bidirNetMap)
			throws RemoteException {
		boolean p = Settings.spt;
		Hashtable<URI, Long> labels = new Hashtable<URI, Long>();
		ArrayList<URI> added = new ArrayList<URI>();
		ArrayList<Link> spt = new ArrayList<Link>();
		URI succ, pred;
		long minLatency;
		added.add(localURI);
		for (int i = 0; i < bidirNetMap.get(0).size(); i++)
			labels.put(bidirNetMap.get(0).get(i).to, 0L);
		if (p)
			System.out.println("SPT computation started");

		// Dijkstra algorithm for spt computation
		while (added.size() < brokers.size()) {
			minLatency = Long.MAX_VALUE;
			succ = null;
			pred = null;
			if (p)
				System.out.println("S={");
			for (URI uri : added)
				if (p)
					System.out.println(" " + uri);
			if (p)
				System.out.println("}");

			// For each node N' in "added"
			for (URI uri : added) {
				if (p)
					System.out.println("Checking outgoing edges from " + uri);

				// For each link L' outgoing from N'...
				for (Link l : getLinksBySource(bidirNetMap, uri)) {

					// ...that belongs to added's outgoing cut
					if (!added.contains(l.to)) {
						if (p)
							System.out.print(" edge " + l.from + " --> " + l.to
									+ "   " + (l.latency + labels.get(l.from))
									+ " (" + l.latency + "+"
									+ labels.get(l.from) + ")");
						if (l.latency + labels.get(l.from) < minLatency) {
							if (p)
								System.out.println(" *");
							minLatency = l.latency + labels.get(l.from);
							succ = l.to;
							pred = l.from;
						} else {
							if (p)
								System.out.println("");
						}
					}
				}
			}
			added.add(succ);
			labels.put(succ, minLatency);
			spt.add(new Link(pred, succ, minLatency));
		}

		// Print computed SPT
		if (p)
			System.out.println("LOCAL SPT COMPUTED");
		for (Link l : spt)
			if (p)
				System.out.println(" " + l.from + " --> " + l.to + "   "
						+ l.latency + "ms");

		// Send your locally computed SPT to all other brokers
		for (BrokerRef br : brokers)
			br.broker.updateRoutingTable(spt, localURI);
	}
}
